package cn.lgwen.candy.runtime.driver;

import cn.lgwen.candy.runtime.operator.SinkOperator;
import cn.lgwen.candy.runtime.queue.Queue;

/**
 * 2019/12/24
 * aven.wu
 * danxieai258@163.com
 */
public class SinkDriver<I> implements Driver {

    private volatile boolean running = true;

    private Queue<I> upstreamQueue;

    private SinkOperator<I> operator;

    private int parallel = 1;

    public SinkDriver(Queue<I> upstreamQueue, SinkOperator<I> operator) {
        this.upstreamQueue = upstreamQueue;
        this.operator = operator;
    }

    public SinkDriver(SinkOperator<I> operator) {
        this.operator = operator;
    }

    public void setUpstreamQueue(Queue<I> upstreamQueue) {
        this.upstreamQueue = upstreamQueue;
    }

    @Override
    public void run() {
        operator.checkpoint(checkpoint());
        while (running) {
            I e = upstreamQueue.take();
            operator.invoke(e);
        }
    }

    @Override
    public void parallel(int parallel) {
        this.parallel = parallel;
    }

    @Override
    public int parallel() {
        return parallel;
    }

    @Override
    public void checkpoint(boolean checkpoint) {

    }

    public void cancel() {
        this.running = false;
    }
}
